
package com.naza.rpc.netty;

import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;

import com.naza.rpc.consumer.balance.manage.ClientManage;
import com.naza.rpc.core.RpcSystemConfig;
import com.naza.rpc.model.ServiceInstance;
import com.naza.rpc.parallel.RpcThreadPool;
import com.naza.rpc.serialize.RpcSerializeProtocol;

import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * RPC服务器配置加载,单列
 * 
 * @author yl
 */
public class RpcServerLoader {

    private static volatile RpcServerLoader rpcServerLoader;

    private ClientManage clientManage = ClientManage.INSTANCE;

    private static final String DELIMITER = RpcSystemConfig.DELIMITER;

    // 序列化方式
    private RpcSerializeProtocol serializeProtocol = RpcSerializeProtocol.JDKSERIALIZE;

    // Java虚拟机可用的处理器数量
    private static final int PARALLEL = RpcSystemConfig.SYSTEM_PROPERTY_PARALLEL * 2;

    // netty nio线程池
    private EventLoopGroup eventLoopGroup = new NioEventLoopGroup(PARALLEL);

    private static int threadNums = RpcSystemConfig.SYSTEM_PROPERTY_THREADPOOL_THREAD_NUMS;

    private static int queueNums = RpcSystemConfig.SYSTEM_PROPERTY_THREADPOOL_QUEUE_NUMS;

    private static ListeningExecutorService threadPoolExecutor = MoreExecutors
            .listeningDecorator((ThreadPoolExecutor) RpcThreadPool.getExecutor(threadNums, queueNums));

    private RpcSendHandler messageSendHandler = null;

    //serviceAddr ,RpcSendInitializeTask
    private Map<String, RpcSendHandler> sendHandlers = new ConcurrentHashMap<>();

    //serviceAddr ,RpcSendInitializeTask
    private Map<String, RpcSendInitializeTask> taskMap = new ConcurrentHashMap<>();

    // 等待netty服务端链路建立通知信号
    private Lock lock = new ReentrantLock();

    private Condition connectStatus = lock.newCondition();

    private Condition handlerStatus = lock.newCondition();

    private RpcServerLoader() {
    }

    // 双重并发锁定
    public static RpcServerLoader getInstance() {
        if (rpcServerLoader == null) {
            synchronized (RpcServerLoader.class) {
                if (rpcServerLoader == null) {
                    rpcServerLoader = new RpcServerLoader();
                }
            }
        }
        return rpcServerLoader;
    }

    public void load(String serverAddress, RpcSerializeProtocol serializeProtocol) {
        if(taskMap.containsKey(serverAddress)){
            return;
        }
        String[] ipAddr = serverAddress.split(RpcServerLoader.DELIMITER);
        if (ipAddr.length == RpcSystemConfig.IPADDR_OPRT_ARRAY_LENGTH) {
            String host = ipAddr[0];
            int port = Integer.parseInt(ipAddr[1]);
            final InetSocketAddress remoteAddr = new InetSocketAddress(host, port);

            System.out.printf(
                    "Naza Client start success!\nip:%s\nport:%d\nprotocol:%s\n\n", host,
                    port, serializeProtocol);
            RpcSendInitializeTask task = new RpcSendInitializeTask(eventLoopGroup, remoteAddr, serializeProtocol);
            ListenableFuture<Boolean> listenableFuture = threadPoolExecutor
                    .submit(task);

            Futures.addCallback(listenableFuture, new FutureCallback<Boolean>() {
                @Override
                public void onSuccess(Boolean result) {
                    try {
                        lock.lock();

                        if (sendHandlers.isEmpty()) {
                            handlerStatus.await();
                        }

                        if (result.equals(Boolean.TRUE) && !sendHandlers.isEmpty()) {
                            connectStatus.signalAll();
                        }
                    } catch (InterruptedException ex) {
                        Logger.getLogger(RpcServerLoader.class.getName()).log(Level.SEVERE, null, ex);
                    } finally {
                        lock.unlock();
                    }
                }

                @Override
                public void onFailure(Throwable t) {
                    t.printStackTrace();
                }
            }, threadPoolExecutor);
            taskMap.put(serverAddress,task);
        }
    }

    public void stopConnectTask(String serverAddress){
        if(taskMap.containsKey(serverAddress)){
            RpcSendInitializeTask task = taskMap.get(serverAddress);
            task.stop();
        }
    }

    public void load(ClientManage clientManage, String serviceName,RpcSerializeProtocol serializeProtocol) {
        List<ServiceInstance> instanceList = clientManage.getByServiceName(serviceName);
        if(null != instanceList){
            for(ServiceInstance instance : instanceList){
                String serverAddress = instance.getIp()+":"+instance.getPort();
                load(serverAddress,serializeProtocol);
            }
        }
    }

    public void setMessageSendHandler(RpcSendHandler messageInHandler, String serviceAddr) {
        try {
            lock.lock();
            //this.messageSendHandler = messageInHandler;
            sendHandlers.put(serviceAddr,messageInHandler);
            // 唤醒所有等待客户端rpc线程
            handlerStatus.signal();
        } finally {
            lock.unlock();
        }
    }

/*    public RpcSendHandler getMessageSendHandler() throws InterruptedException {
        try {
            lock.lock();
            // Netty服务端链路没有建立完毕之前，先挂起等待
            if (messageSendHandler == null) {
                connectStatus.await();
            }
            return messageSendHandler;
        } finally {
            lock.unlock();
        }
    }*/

    public RpcSendHandler getMessageSendHandler(String serviceAddr) throws InterruptedException {
        try {
            lock.lock();
            // Netty服务端链路没有建立完毕之前，先挂起等待
            if (sendHandlers.isEmpty()) {
                connectStatus.await();
            }
            if(!sendHandlers.containsKey(serviceAddr)){
                //新加入的实例，连接
                load(serviceAddr, RpcSendExecutor.getInstance().getProtocol());
            }
            return sendHandlers.get(serviceAddr);
        } finally {
            lock.unlock();
        }
    }

    public void unLoad() {
        for(Map.Entry<String, RpcSendHandler> entry : sendHandlers.entrySet()){
            entry.getValue().close();
        }
        //messageSendHandler.close();
        threadPoolExecutor.shutdown();
        eventLoopGroup.shutdownGracefully();
    }

    public void setSerializeProtocol(RpcSerializeProtocol serializeProtocol) {
        this.serializeProtocol = serializeProtocol;
    }
}
